package com.aispeech.asr.apis.ssc.service;

import static org.springframework.data.mongodb.core.query.Criteria.where;
import static org.springframework.data.mongodb.core.query.Update.update;

import java.util.List;

import javax.annotation.PostConstruct;

import org.springframework.context.ApplicationContext;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.FindAndModifyOptions;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service;

import com.aispeech.asr.apis.ssc.config.AppConfig;
import com.aispeech.asr.apis.ssc.constants.Progress;
import com.aispeech.asr.apis.ssc.data.domain.AudioInfo;
import com.aispeech.asr.apis.ssc.data.domain.TaskDoc;
import com.aispeech.asr.apis.ssc.event.SscEvent;
import com.aispeech.asr.apis.ssc.metrics.Prometheus;
import com.aispeech.asr.comm.data.pojo.PauseStatusDoc;
import com.aispeech.asr.comm.enums.EAudioType;
import com.aispeech.asr.comm.enums.ErrorCode;

import lombok.extern.slf4j.Slf4j;

@Service
@Slf4j
public class TaskSchdService {
    private final AppConfig config;
    private final MongoOperations mongoOperations;
    private final ProgressManager progressManager;
    private final ApplicationContext applicationContext;

    private String nodeName;

    public TaskSchdService(AppConfig config, MongoOperations mongoOperations, ProgressManager progressManager, ApplicationContext applicationContext) {
        this.config = config;
        this.mongoOperations = mongoOperations;
        this.progressManager = progressManager;
        this.applicationContext = applicationContext;
    }

    public void run() {
        List<String> taskIdListWIP = progressManager.taskIdListInProgress();

        if (taskIdListWIP.size() > 0)
            log.info("WIP: {}, max: {}", progressManager.taskInfoListInProgress(), config.getNWorkers());

        if (taskIdListWIP.size() < config.getNWorkers()) {
            try {
                TaskDoc taskDoc = pickOneTaskFromMongo(taskIdListWIP);

//                log.info("[ run ] taskIdListWIP: {}, taskDoc： {}", taskIdListWIP, taskDoc);

                if (!(taskDoc == null || taskIdListWIP.contains(taskDoc.getId()))
                        && nodeName.equals(taskDoc.getHostname())
                        //&& !isPaused(taskDoc)
                        ) {
                    if (taskDoc.getProgress() == Progress.IDLE)
                        progressManager.syncToMongo(taskDoc, Progress.PREPARE);
                    else {
                        progressManager.getTasksInProgress().add(taskDoc);
                    }
                    // 本机worker线程之间全靠 taskIdListWIP 去重
                    // 多台executor机器，靠pick里的hostname抢占机制
                    if (taskDoc.getProductId() == null)
                        taskDoc.setProductId((String) taskDoc.getRequest().getOrDefault("appId", "unknown"));

                    Object objDuration = taskDoc.getRequest().getOrDefault("duration", 0);
                    if (objDuration.getClass() == Double.class) {
                        long duration = ((Double) objDuration).longValue();
                        taskDoc.getRequest().put("duration", duration);
                    }

                    doTask(taskDoc);
                } else {
                    if (taskDoc != null)
                        log.error("[ unexpected ]: id: {}, hostname: {}", taskDoc.getId(), taskDoc.getHostname());
                }
            } catch (Exception ex) {
                log.warn("run cron task failed. {}", ex.getMessage());
            }
        }
    }

    private boolean isPaused(TaskDoc taskDoc) {
        boolean paused = false;

        Query q = Query.query(
                where("lang").is(taskDoc.getRequest().get("lang"))
                        .and("rate").is(taskDoc.getRequest().get("sampleRate")));

        PauseStatusDoc pauseStatusDoc = mongoOperations.findOne(q, PauseStatusDoc.class);
        if (pauseStatusDoc != null) {
            paused = pauseStatusDoc.isPaused();

            if (paused) {
                log.warn("paused. {}", q);
            }
        }

        return paused;
    }

    private void doTask(TaskDoc taskDoc) {
        SscEvent event = new SscEvent(this, taskDoc);
        EAudioType audioType = EAudioType.fromKey((String) taskDoc.getRequest().get("audioType"));
        if (audioType == null) {
            progressManager.failed(
                    event,
                    ErrorCode.ERR_AUDIO_CREATE_FAILED.errid,
                    String.format("irregular audio type: %s", taskDoc.getRequest().get("audioType"))
            );
        } else {
        	event.setAudioInfo(new AudioInfo(audioType, taskDoc.getId(),
                    (int) taskDoc.getRequest().getOrDefault("sampleRate", taskDoc.getRequest().get("sampleRates")),
                    (int) taskDoc.getRequest().getOrDefault("sampleBytes", 2),
                    (int) taskDoc.getRequest().getOrDefault("channel", 1)
            ));
            applicationContext.publishEvent(event);
        }
    }

    private TaskDoc pickOneTaskFromMongo(List<String> taskIdListWIP) {
        // 条件1. 必须不能是已经在本机进行中（ 不在taskIdListWIP队列）
        Criteria notInWIP = where("_id").nin(taskIdListWIP);

        // 条件2. 选择一个没有hostname，且进度为IDLE的
        Criteria idleAndNonHost = where("progress").is(Progress.IDLE).and("hostname").is(null);

        // 条件3. 或者选上次重启前，hostname为本机，且没有执行完的
        Criteria brokenWhenRestart = where("hostname").is(nodeName)
                .and("progress").lt(Progress.FUSED)
                .and("error").is(null);

        // 条件1 and (条件2 or 条件3)
        Query query = Query.query(
                new Criteria().andOperator(
                        notInWIP,
                        new Criteria().orOperator(
                                idleAndNonHost,
                                brokenWhenRestart
                        )
                ))
                .with(Sort.by(
                        Sort.Order.desc("priority"),
                        Sort.Order.asc("update_t")
                ))
                .limit(1);

        return mongoOperations.findAndModify(
                query,
                update("hostname", nodeName),        // **** 全靠本次pick，和其它主机竞争 ****
                new FindAndModifyOptions().returnNew(true),
                TaskDoc.class
        );
    }

    @PostConstruct
    public void init() {
        nodeName = Prometheus.NodeName();
    }
}
